mod shared;
mod opus_ogg;
mod conf_bridge;

use crate::shared::MAX_UDP_BUFFER_SIZE;
use crate::opus_ogg::OpusOggReader;
use crate::conf_bridge::{ConfBridge, ConfBridgeChannel};

use bitvec::prelude::*;
use conf_bridge::TransportPacket;
use jarust::jaconfig::JaConfig;
use jarust::jaconfig::TransportType;
use jarust::japrotocol::EstablishmentProtocol;
use jarust::japrotocol::RTP;
use jarust_plugins::audio_bridge::messages::AudioBridgeCreateOptions;
use jarust_plugins::audio_bridge::messages::AudioBridgeJoinOptions;
use jarust_plugins::audio_bridge::AudioBridge;
use tracing_subscriber::EnvFilter;
use bitvec::prelude::*;
use clap::{Arg, ArgAction};
use flexi_logger;
use hex::{self, FromHex};
use log;
use std::time::Duration;
use std::{
    collections::VecDeque,
    ffi::c_int,
    fs::File,
    io::{BufRead, BufReader},
    net::SocketAddr,
    sync::{mpsc::channel, Arc, Mutex},
};
use tokio::{
    self,
    sync::{mpsc, Notify, Semaphore},
};

struct RtpPacket {
    data: Vec<u8>,
}

fn main() {
    const TIME_FORMAT: &str = "%Y-%m-%d %H:%M:%S.%3f";
    let logger = flexi_logger::Logger::try_with_env()
        .unwrap()
        .format(|buf, now, record| {
            write!(
                buf,
                "{} {:6}{} - {}",
                now.format(TIME_FORMAT),
                record.level(),
                record.target(),
                record.args()
            )
        })
        .start()
        .unwrap();

    let opts = clap::Command::new("test mixer")
        .version("0.1")
        .author("Wang Hoi <wanghoi@126.com>")
        .about("Does awesome things")
        .arg(
            Arg::new("local")
                .long("local")
                .value_name("LOCAL")
                .help("Sets local address")
                .default_value("127.0.0.1")
                .action(ArgAction::Set),
        )
        .arg(
            Arg::new("server")
                .short('s')
                .long("server")
                .value_name("SERVER")
                .help("Sets Janus server address")
                .default_value("192.168.6.53")
                .action(ArgAction::Set),
        )
        .arg(
            Arg::new("num-rooms")
                .short('n')
                .long("num-rooms")
                .value_name("NUM")
                .help("Sets number to mixer rooms")
                .default_value("1")
                .action(ArgAction::Set),
        )
        .arg(
            Arg::new("num-participants")
                .short('p')
                .long("num-participants")
                .value_name("NUM")
                .help("Sets per-room participants")
                .default_value("9")
                .action(ArgAction::Set),
        )
        .arg(
            Arg::new("rtp-file")
                .short('f')
                .long("rtp-file")
                .value_name("FILE")
                .help("Sets rtp file")
                .default_value("opus_rtp_udp.txt")
                .action(ArgAction::Set),
        )
        .arg(
            Arg::new("threads")
                .short('t')
                .long("threads")
                .value_name("NUM")
                .help("Sets worker thread")
                .default_value("1")
                .action(ArgAction::Set),
        )
        .get_matches();

    let local_ip = opts
        .get_one::<String>("local")
        .expect("local ip parse error.");
    let server_ip = opts
        .get_one::<String>("server")
        .expect("server ip parse error.");
    let num_rooms = opts
        .get_one::<String>("num-rooms")
        .unwrap()
        .parse::<i32>()
        .expect("num-rooms parse error.");
    let num_participants = opts
        .get_one::<String>("num-participants")
        .unwrap()
        .parse::<i32>()
        .expect("num-participants parse error.");
    let rtp_file = opts
        .get_one::<String>("rtp-file")
        .expect("rtp-file parse error.");
    let threads = opts
        .get_one::<String>("threads")
        .unwrap()
        .parse::<usize>()
        .expect("threads parse error.");
    let rt = if true/*threads == 0 || threads > 1*/ {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(threads)
            .enable_all()
            .build()
            .expect("Init thread runtime error")
    } else {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("Init thread runtime error")
    };

    log::warn!(
        "start janus testing: [{}->{}], {} worker-threads, {} rooms, {} per-room participants, rtp-file=[{}]...",
        local_ip,
        server_ip,
        threads,
        num_rooms,
        num_participants,
        rtp_file
    );

    rt.block_on(async_main(local_ip.clone(), server_ip.clone(), num_rooms, num_participants, rtp_file));
}

async fn async_main(local_ip:String, server_ip: String, num_rooms: i32, num_participants: i32, rtp_file: &str) {
    let rtp_packets = Arc::new(load_rtp_data(rtp_file).unwrap_or_default());

    for i in 0..num_rooms {
        let cid = format!("1{:05}", i);
        let conf_bridge = Arc::new(ConfBridge::new(&local_ip, &server_ip, &cid, String::new()));
        let conf_id = format!("c{}", i);
        let mut participant_ids = Vec::new();
        for j in 0..num_participants {
            let participant_id = format!("2{:05}{:05}", i, j);
            participant_ids.push(participant_id.clone());
            let (tx, rx) = mpsc::channel(50);
            let local_payload_type = 96u8;
            let local_ssrc = (i * 1000 + j) as u32;
            let remote_payload_type = 96u8;
            let ssrc: Option<u32> = Some(j as u32);
            let chan = conf_bridge
                .clone()
                .create_channel(
                    &participant_id,
                    tx,
                    local_payload_type,
                    local_ssrc,
                    remote_payload_type,
                    ssrc,
                )
                .await
                .expect("create_channel failed");
            tokio::spawn(conf_participant_loop(
                conf_id.clone(),
                participant_id,
                conf_bridge.clone(),
                j,
                chan,
                rx,
                ssrc,
                rtp_packets.clone(),
            ));
            // tokio::time::sleep(Duration::from_millis(100)).await;
        }
        tokio::spawn(conf_manage_loop(conf_id, participant_ids, conf_bridge));
    }

    tokio::time::sleep(tokio::time::Duration::from_secs(3600)).await;
}

fn init_rtp_header(header: &mut [u8], payload: u8, ssrc: u32) {
    let bs = BitSlice::<_, Msb0>::from_slice_mut(header);
    bs.set(0, true);
    bs[8..16].store_be(payload);
    bs[64..96].store_be(ssrc);
}

fn update_rtp_header(header: &mut [u8], seq: u16, timestamp: u32) {
    let bs = BitSlice::<_, Msb0>::from_slice_mut(header);
    bs[16..32].store_be(seq);
    bs[32..64].store_be(timestamp);
}

async fn conf_manage_loop(
    conf_id: String,
    participant_ids: Vec<String>,
    conf_bridge: Arc<ConfBridge>,
) {
    // let end_time = tokio::time::Instant::now() + tokio::time::Duration::from_millis(100);
    // tokio::time::sleep_until(end_time).await;
    // log::info!("cid {} end", conf_id);
    // drop(conf_bridge);
}

async fn conf_participant_loop(
    conf_id: String,
    participant_id: String,
    conf_bridge: Arc<ConfBridge>,
    channel_id: i32,
    chan: ConfBridgeChannel,
    mut rx: mpsc::Receiver<TransportPacket>,
    ssrc: Option<u32>,
    rtp_packets: Arc<Vec<RtpPacket>>,
) {
    let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(20));
    let mut rtp_buf = [0u8; MAX_UDP_BUFFER_SIZE];
    let mut next_index = 0usize;
    let mut rtp_pkt_size: usize;
    let end_time = tokio::time::Instant::now() + tokio::time::Duration::from_millis(100);
    loop {
        tokio::select! {
            // _ = tokio::time::sleep_until(end_time) => {
                // log::debug!("cid {} pid {} break", conf_id, participant_id);
                // break;
            // }
            pkt = rx.recv() => {},
            _ = (&mut interval).tick() => {
                if rtp_packets.len() > 0 {
                    rtp_pkt_size = rtp_packets[next_index % rtp_packets.len()].data.len();
                    (&mut rtp_buf[0..rtp_pkt_size]).copy_from_slice(&rtp_packets[next_index].data[0..rtp_pkt_size]);
                    init_rtp_header(&mut rtp_buf, 96, ssrc.unwrap_or(channel_id as u32));
                    let seq: u16 = (next_index & 0xffff) as u16;
                    let timestamp = ((next_index * 960) & 0xffffffff) as u32;
                    update_rtp_header(&mut rtp_buf, seq, timestamp);
                    next_index = next_index + 1;
                } else {
                    init_rtp_header(&mut rtp_buf, 96, ssrc.unwrap_or(channel_id as u32));
                    rtp_pkt_size = 120;
                }
                //log::info!("{} recv rtp size {}", participant_id, rtp_pkt_size);
                chan.received_rtp(&rtp_buf[..rtp_pkt_size]);
            }
            else => {
                break;
            }
        }
    }
}

fn load_rtp_data(fname: &str) -> std::io::Result<Vec<RtpPacket>> {
    let file = File::open(fname)?;
    let reader = BufReader::new(file);
    let mut rtp_packets = vec![];
    for line in reader.lines() {
        if let Ok(raw) = hex::decode(line?) {
            rtp_packets.push(RtpPacket {
                data: raw[0..].to_vec(),
            });
            //log::info!("pkt len {}", rtp_packets.last().as_ref().unwrap().data.len());
        }
    }
    log::info!("read {} packets", rtp_packets.len());
    Ok(rtp_packets)
}

/*
#[tokio::main(flavor = "current_thread")]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::from_default_env()
                .add_directive("info".parse()?)
                .add_directive("jarust=debug".parse()?),
        )
        .init();
    let timeout = std::time::Duration::from_secs(10);
    tracing::info!("Start polling events...");
    let config = JaConfig::builder().url("ws://localhost:8188/ws").build();
    let mut connection = jarust::connect(config, TransportType::Ws).await?;
    let session = connection.create(10).await?;
    let (handle, mut event_receiver) = session.attach_audio_bridge().await?;

    let room = 2345u64;
    let result = handle.list(timeout).await?;
    tracing::info!("Rooms {:#?}", result);

    let options = AudioBridgeCreateOptions {
        room: Some(room),
        allow_rtp_participants: Some(true),
        ..Default::default()
    };
    let _result = handle.create_room_with_config(options, timeout).await?;

    let options = AudioBridgeJoinOptions {
        id: 1.into(),
        ..Default::default()
    };
    let protocol = EstablishmentProtocol::RTP(RTP {
        ip: String::from("192.168.6.53"),
        port: 12340,
        payload_type: 96.into(),
        audiolevel_ext: None,
        fec: None,
    });
    let _result = handle
        .join_room(room, options.clone(), protocol.clone().into(), timeout)
        .await
        .map_err(|e| {
            tracing::error!("join_room error: {}", e);
            e
        })?;

    // let result = handle.list_participants(room, timeout).await.map_err(|e| {
    //     tracing::error!("list_participants error: {}", e);
    //     e
    // })?;
    // tracing::info!("Room participants {:#?}", result);

    loop {
        while let Some(event) = event_receiver.recv().await {
            tracing::info!("Joined Room {}, {:#?}", room, event);
        }
    }
    Ok(())
}

struct OpusHeader {
    channel_count: usize,
    sample_rate: usize,
}

fn load_opus_data(fname: &str) -> std::io::Result<(OpusHeader, Vec<Vec<u8>>)> {
    let file = File::open(fname)?;
    let mut reader = OpusOggReader::new(file).expect("create opus reader failed");
    eprintln!(
        "channel_count={}, sample_rate={}",
        reader.channel_count, reader.sample_rate
    );
    let hdr = OpusHeader {
        channel_count: reader.channel_count,
        sample_rate: reader.sample_rate,
    };
    let mut out = Vec::new();
    while let Some(pkt) = reader.read_packet() {
        // eprintln!("frame len={}", pkt.len());
        out.push(pkt);
    }
    Ok((hdr, out))
}

struct RtpPacket {
    data: Vec<u8>,
}

fn load_rtp_data(fname: &str) -> std::io::Result<Vec<RtpPacket>> {
    let file = File::open(fname)?;
    let reader = BufReader::new(file);
    let mut rtp_packets = vec![];
    for line in reader.lines() {
        if let Ok(raw) = hex::decode(line?) {
            rtp_packets.push(RtpPacket {
                data: raw[0..].to_vec(),
            });
            //log::info!("pkt len {}", rtp_packets.last().as_ref().unwrap().data.len());
        }
    }
    log::info!("read {} packets", rtp_packets.len());
    Ok(rtp_packets)
}

fn init_rtp_header(header: &mut [u8], payload: u8, ssrc: u32) {
    let bs = BitSlice::<_, Msb0>::from_slice_mut(header);
    bs.set(0, true);
    bs[8..16].store_be(payload);
    bs[64..96].store_be(ssrc);
}

fn update_rtp_header(header: &mut [u8], seq: u16, timestamp: u32) {
    let bs = BitSlice::<_, Msb0>::from_slice_mut(header);
    bs[16..32].store_be(seq);
    bs[32..64].store_be(timestamp);
}
*/